package com.leilei.time.slide;

import com.leilei.Location;
import com.leilei.count.MyLocationSource;
import com.leilei.time.tumble.TumbleWindowFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.util.List;

/**
 * @author lei
 * @version 1.0
 * @date 2021/5/24 21:13
 * @desc
 */
public class FlinkWindow_TimeWindow_Sliding {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        DataStreamSource<Location> stream = env.addSource(new MyLocationSource());
        WindowedStream<Location, Integer, TimeWindow> windowStream = stream
                .keyBy(Location::getVehicleId)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)));
        SingleOutputStreamOperator<List<Location>> alarmSource = windowStream.apply(new SlidingWindowFunction());
        alarmSource.print();
        env.execute("slide-time-window");
    }
}
